package com.rrd.dw.mr.gzip.bhzx;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Pattern;

import org.apache.commons.math.ode.ODEIntegrator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.rrd.dw.mr.key.IntPair;
import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.HadoopUtils;
import com.rrd.dw.utils.MD5Util;

public class BHZXTest  extends Configured implements Tool{
	public static final String CRT_FLAG = "\001";
    public static final Pattern CRT_PATTERN = Pattern.compile(CRT_FLAG);
    
	public static class D3Maper extends Mapper<LongWritable, Text, IntPair, Text> {
		 private IntPair ekey = new IntPair();
		private String line = "";
		private String arr[]=null;
		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, IntPair, Text>.Context context)
				throws IOException, InterruptedException {
			if(value==null){
				return;
			}
			line=value.toString();
			arr=CRT_PATTERN.split(line);
			if(arr==null || arr.length<17){
				return;
			}
		
			
			ekey.set(arr[0], Integer.parseInt(arr[7]));
			context.write(ekey, value);
		}
	}

	public static class D3Reduce extends Reducer<IntPair, Text, NullWritable, Text> {
		private static NullWritable key_null = NullWritable.get();
		private String arr[]=null;
		private Map<String,String> map = new HashMap<String,String>(20);
		private Text val_Text = new Text();
		double act_repay_amt=0; 
		double overdue_amt=0;
		int old_days=0;
		@Override
		protected void setup(Reducer<IntPair, Text, NullWritable, Text>.Context context)
				throws IOException, InterruptedException {
			act_repay_amt=0;
			overdue_amt=0;
		}
		@Override
		protected void reduce(IntPair key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			map.clear();
			double loan_amt=0;
//			double derate_amt=0;
			double planned_payment=0;
		    double due_loan_amount=0;
			act_repay_amt=0; 
			overdue_amt=0;
			String min_due_date="";
			int i=0;
			for(Text t :values){
				
				arr = CRT_PATTERN.split(t.toString());
				map.put("loan_key", arr[0]);
				map.put("loan_id", arr[1]);
				map.put("loan_payoff_time",arr[2] );
				map.put("real_repayment", arr[3]);
				map.put("loan_amt", arr[4]);
				map.put("loan_status", arr[5]);
				map.put("repay_plan_key", arr[6]);
				map.put("term_no", arr[7]);
				map.put("target_repayment_date", arr[8]);
				map.put("payoff_time", arr[9]);
				map.put("repay_type", arr[10]);
				map.put("principal", arr[11]);
				map.put("planned_payment", arr[12]);
				map.put("create_time", arr[13]);
				map.put("loan_platform", arr[14]);
				map.put("name", arr[15]);
				map.put("pid", arr[16]);
				map.put("mobile", arr[17]);
				map.put("od_days", arr[18]);
				map.put("derate_amt", arr[19]);
				map.put("next_due_date", arr[20]);  //下一期应还款日期
				map.put("last_due_date", arr[21]);  //最后一期
				map.put("dt", arr[23]);  //观察日
				map.put("loan_platform", arr[24]);  //进件平台
				loan_amt=Double.parseDouble(map.get("loan_amt"));
				planned_payment=Double.parseDouble(map.get("planned_payment"));
				 
//				derate_amt=Double.parseDouble(map.get("derate_amt"));  
				
				map.put("op_code", "A"); 
				
				
			    double  real_repayment=Double.parseDouble(map.get("real_repayment"));
				String confirm_at="";
				String uploadTs="";
				if(map.get("payoff_time").equals("") || map.get("payoff_time").substring(0,10).compareTo(map.get("target_repayment_date").substring(0,10))>0  ){
					confirm_at=DateUtil.formateDateStr(DateUtil.addDay(DateUtil.formateToDate(map.get("target_repayment_date"),"yyyy-MM-dd"), 1));
					if(min_due_date.equals("")){
						min_due_date=map.get("target_repayment_date");
					}
				}else{
					confirm_at=map.get("payoff_time"); 
					//截止当前已经没有逾期	 
					min_due_date="";
				}
				uploadTs=DateUtil.getCurrentDateTime();
				 
				//应还款次日上报
				//计算逾期天数 
				 
				if(min_due_date.equals("")){
					old_days=0;
				}else{
					old_days=DateUtil.date_diff(confirm_at, min_due_date);
				}
				if(old_days<=0){
					
					due_loan_amount=due_loan_amount+(real_repayment<planned_payment?real_repayment:planned_payment);
				}else{
					due_loan_amount=due_loan_amount+planned_payment;
				}
				
				overdue_amt=overdue_amt+(old_days>0?planned_payment:0);
				//act_repay_amt 主要用来计算剩余金额，不是征信指标值
				act_repay_amt=act_repay_amt+(old_days>0?0:planned_payment);
				i++;
				map.put("req_id", MD5Util.getMd5(map.get("repay_plan_key")+uploadTs+i));
				map.put("uploadTs", uploadTs);
				map.put("term_status",old_days>0?"overdue":"normal");
				map.put("real_repayment_date",old_days>0?"":map.get("payoff_time")); //实际还款时间
				map.put("target_repayment",String.valueOf(due_loan_amount));
				map.put("real_repayment", old_days>0?"0":String.valueOf(real_repayment));	
				map.put("overdue_status", old_days>0?"D"+old_days:"");
				map.put("status_confirm_at",confirm_at);
				
				map.put("remaining_amount", String.valueOf(loan_amt-act_repay_amt));
				map.put("act_repay_amt", String.valueOf(act_repay_amt));
				if(map.get("last_due_date").equals(map.get("target_repayment_date")) &&  old_days<=0){
					map.put("loan_status", "3");
					overdue_amt=0;
				}else if(old_days>0){
					map.put("loan_status", "2");
				}else{
					map.put("loan_status", "1");
				} 
				map.put("overdue_amount", String.valueOf(overdue_amt)); //当 前 逾期总额
				map.put("repay_type", old_days+","+map.get("last_due_date")+","+map.get("target_repayment_date")+","+confirm_at+","+map.get("dt")+","+map.get("real_repayment"));
				 
				val_Text.set(processToString(map));
				context.write(key_null, val_Text);
				//判断如果本次是支付
				if(old_days<=0){
					due_loan_amount=0;
				}
//				处理逾期情况,并且是在下一期到期之前，已还
				
				if(old_days>0 && !map.get("payoff_time").equals("") && (map.get("next_due_date").compareTo(map.get("payoff_time"))>0
						|| map.get("last_due_date").equals(map.get("target_repayment_date"))))
				{
					confirm_at=map.get("payoff_time");
//					uploadTs=map.get("payoff_time");
					min_due_date="";
					old_days=0;
					act_repay_amt=act_repay_amt+ planned_payment;
					overdue_amt=overdue_amt-planned_payment;
					i++;
					map.put("req_id", MD5Util.getMd5(map.get("repay_plan_key")+uploadTs+i));
					map.put("uploadTs", uploadTs);
					map.put("term_status","normal");
					map.put("real_repayment_date",map.get("payoff_time"));
					map.put("target_repayment",String.valueOf(real_repayment));
					map.put("real_repayment", String.valueOf(real_repayment));	
					map.put("overdue_status", "");
					map.put("status_confirm_at", confirm_at);
					
					map.put("remaining_amount", String.valueOf(Double.parseDouble(map.get("loan_amt"))-act_repay_amt));
					if(map.get("last_due_date").equals(map.get("target_repayment_date")) &&  !map.get("loan_payoff_time").equals("")){
						map.put("loan_status", "3");
						overdue_amt=0;
					}else{
						map.put("loan_status", "1");
						overdue_amt=0;
					} 
					map.put("overdue_amount", String.valueOf(overdue_amt));
					map.put("act_repay_amt", String.valueOf(act_repay_amt));
					map.put("repay_type", old_days+","+map.get("last_due_date")+","+map.get("target_repayment_date")+","+confirm_at+","+map.get("dt")+","+map.get("real_repayment"));
					val_Text.set(processToString(map));
					context.write(key_null, val_Text);
					due_loan_amount=0;
				}
				
				//最后一期，整笔逾期30天以上，并且没有还清,每30天上报一次
				
				if(old_days>0 && map.get("last_due_date").equals(map.get("target_repayment_date"))){
					confirm_at=DateUtil.formateDateStr(DateUtil.addDay(DateUtil.formateToDate(confirm_at), 30));
					while(confirm_at.compareTo(map.get("dt"))<=0){
						old_days=DateUtil.date_diff(confirm_at, min_due_date);
						i++;
						map.put("req_id", MD5Util.getMd5(map.get("repay_plan_key")+uploadTs+i));
						map.put("uploadTs", uploadTs);
						map.put("term_status","overdue");
						map.put("real_repayment_date",""); //实际还款时间
						map.put("target_repayment",String.valueOf(due_loan_amount));
						map.put("real_repayment", "0");	
						map.put("overdue_status", "D"+old_days);
						map.put("status_confirm_at",confirm_at);
						map.put("overdue_amount", String.valueOf(overdue_amt)); //当 前 逾期总额
						map.put("remaining_amount", String.valueOf(loan_amt-act_repay_amt));
						map.put("loan_status", "2");
						map.put("act_repay_amt", String.valueOf(act_repay_amt));
						map.put("repay_type", old_days+","+map.get("last_due_date")+","+map.get("target_repayment_date")+","+confirm_at+","+map.get("dt"));
						val_Text.set(processToString(map));
						context.write(key_null, val_Text);
						confirm_at=DateUtil.formateDateStr(DateUtil.addDay(DateUtil.formateToDate(confirm_at), 30));
					}
				}
			}
			 
			
		}
		
	}
	private static String processToString(Map<String,String> map){
		StringBuilder sb = new StringBuilder();
//		sb.append(MD5Util.getMd5(map.get("repay_plan_key")+map.get("status_confirm_at"))).append(CRT_FLAG);
		sb.append(map.get("loan_key")).append(CRT_FLAG);
		sb.append(map.get("repay_plan_key")).append(CRT_FLAG);
		sb.append(map.get("req_id")).append(CRT_FLAG);
		sb.append("A").append(CRT_FLAG); //op_code
		sb.append(map.get("uploadTs")).append(CRT_FLAG);
		sb.append(map.get("loan_id")).append(CRT_FLAG);
		sb.append(map.get("name")).append(CRT_FLAG);
		sb.append(map.get("pid")).append(CRT_FLAG);
		sb.append(map.get("mobile")).append(CRT_FLAG);
		sb.append(map.get("term_no")).append(CRT_FLAG);
		sb.append(map.get("term_status")).append(CRT_FLAG);
		sb.append(map.get("target_repayment_date")).append(CRT_FLAG);
		sb.append(map.get("real_repayment_date")).append(CRT_FLAG);
		sb.append(map.get("planned_payment")).append(CRT_FLAG);
		sb.append(map.get("target_repayment")).append(CRT_FLAG);
		sb.append(map.get("real_repayment")).append(CRT_FLAG);
		sb.append(map.get("overdue_status")).append(CRT_FLAG);
		sb.append(map.get("status_confirm_at")).append(CRT_FLAG);
		sb.append(map.get("overdue_amount")).append(CRT_FLAG);
		sb.append(map.get("remaining_amount")).append(CRT_FLAG);
		sb.append(map.get("loan_status")).append(CRT_FLAG);
		sb.append(map.get("act_repay_amt")).append(CRT_FLAG);
		sb.append(map.get("od_days")).append(CRT_FLAG);
		sb.append(map.get("payoff_time")).append(CRT_FLAG);
		sb.append(map.get("last_due_date")).append(CRT_FLAG);
		sb.append(map.get("loan_amt")).append(CRT_FLAG);
		sb.append(map.get("loan_payoff_time")).append(CRT_FLAG);
		sb.append(map.get("loan_platform")).append(CRT_FLAG);
		sb.append(map.get("loan_key")).append(CRT_FLAG);
		sb.append(map.get("repay_type"));
		return sb.toString();
		 
	}
	/*
     * 自定义分区函数类FirstPartitioner，根据 IntPair中的first实现分区
     */
    public static class FirstPartitioner extends Partitioner<IntPair, Text>{
        @Override
        public int getPartition(IntPair key, Text value,int numPartitions){ 
            return Math.abs(key.getFirst().hashCode() * 127) % numPartitions;
        }
    }
    
    /*
     * 自定义GroupingComparator类，实现分区内的数据分组
     */
    @SuppressWarnings("rawtypes")
    public static class GroupingComparator extends WritableComparator{
        protected GroupingComparator(){
            super(IntPair.class, true);
        }
        
        @Override
        public int compare(WritableComparable w1, WritableComparable w2){
            IntPair ip1 = (IntPair) w1;
            IntPair ip2 = (IntPair) w2;
            String l = ip1.getFirst();
            String r = ip2.getFirst();
            return l.equals(r) ? 0 : (l.compareTo(r)<0? -1 : 1);
        }
    }
	@Override
	public int run(String[] args) throws Exception {

		 Configuration conf = getConf();
	        Job job = Job.getInstance( conf);
	        conf.set("mapreduce.output.fileoutputformat.compress", "true");
	        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
	        conf.set("mapreduce.output.fileoutputformat.compress.codec",
	            "org.apache.hadoop.io.compress.GzipCodec");
	        conf.set("io.compression.codecs",
	            "org.apache.hadoop.io.compress.GzipCodec");
	         
	        
	        String inputDir = conf.get("param.input.path");
	         String outDir = conf.get("param.out.path");
	         if(inputDir==null || inputDir.equals("")){
	        	 System.out.println("请输入输入路径：param.input.path");
	        	 System.exit(0);
	         }
	         if(outDir==null || outDir.equals("")){
	        	 System.out.println("请输入输出路径：param.out.path");
	        	 System.exit(0);
	         }
	        System.out.println("param.input.path=" + inputDir);
	        System.out.println("param.out.path=" + outDir);
	        Path outpath = new Path(outDir);
	        HadoopUtils.delete(conf, outpath);  
	        TextInputFormat.setInputPaths(job, new Path(inputDir));
	        TextOutputFormat.setOutputPath(job, new Path(outDir));
	        FileOutputFormat.setCompressOutput(job, true);
	        FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class); //设置压缩格式
	        job.setMapperClass(D3Maper.class);
	        job.setReducerClass(D3Reduce.class);
	        job.setPartitionerClass(FirstPartitioner.class);
	        job.setGroupingComparatorClass(GroupingComparator.class);
	        job.setOutputKeyClass(Text.class);
	        job.setMapOutputKeyClass(IntPair.class);
	        job.setJarByClass(this.getClass());
	        
 
	        return job.waitForCompletion(true) ? 0 : 1;
		 
	}
	public static void main(String[] args) throws Exception {
		 int res = ToolRunner.run(new BHZXTest(), args);
	        System.exit(res);
		
	}
	
}
